package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.WarnAmplitudeBean;
import cn.itcast.bean.WarnBaseBean;
import cn.itcast.inter.ProcessDataCepInterface;
import cn.itcast.mail.MailSend;
import cn.itcast.util.RedisUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import redis.clients.jedis.JedisCluster;

import java.math.BigDecimal;
import java.util.List;
import java.util.Map;

/**
 * Created by JiaYang Xu
 * Date 2021/7/8 16:26
 * Description
 */
public class AmplitudeTask implements ProcessDataCepInterface {
    @Override
    public void process(DataStream<CleanBean> waterData, StreamExecutionEnvironment env) {
        //计算公式：振幅： (max(最高点位)-min(最低点位))/期初前收盘点位*100% ,secCode,eventTime
        /**
         * 1.数据转换
         * 2.初始化表执行环境
         * 3.注册表（流）
         * 4.sql执行
         * 5.表转流
         * 6.模式匹配,比较阀值
         * 7.查询数据
         * 8.发送告警邮件
         */
//        1.数据转换
        SingleOutputStreamOperator<WarnBaseBean> warnData = waterData.map(new MapFunction<CleanBean, WarnBaseBean>() {
            @Override
            public WarnBaseBean map(CleanBean value) throws Exception {
                WarnBaseBean warnBean = new WarnBaseBean(
                        value.getSecCode(),
                        value.getPreClosePrice(),
                        value.getMaxPrice(),
                        value.getMinPrice(),
                        value.getTradePrice(),
                        value.getEventTime());
                return warnBean;
            }
        });
//        初始化表执行环境
        StreamTableEnvironment tableEnvironment = TableEnvironment.getTableEnvironment(env);
        //注册表（流）
        //振幅： (max(最高点位)-min(最低点位))/期初前收盘点位*100%
        //使用 registerDataStream 注册表，并分别指定字段，还要指定rowtime字段secCode,preClosePrice,highPrice,lowPrice,eventTime.rowtime
        tableEnvironment.registerDataStream("tbl", warnData, "secCode,preClosePrice,highPrice,lowPrice,eventTime.rowtime");
        //sql执行
        String sql = "select secCode,preClosePrice,round((max(highPrice)-min(lowPrice))/preClosePrice,2) as amplitude " +
                " from tbl group by secCode,preClosePrice,tumble(eventTime, interval '5' second)";
        Table table = tableEnvironment.sqlQuery(sql);
//        表转流
        DataStream<WarnAmplitudeBean> resDate = tableEnvironment.toAppendStream(table, TypeInformation.of(WarnAmplitudeBean.class));
        resDate.print("sql查询数据：");
        //6.模式匹配,比较阀值
        //先获取阀值
        JedisCluster jedisCluster = RedisUtil.getJedisCluster();
        BigDecimal amplitudeThreshold = new BigDecimal(jedisCluster.hget("quot", "amplitude"));
        //定义匹配规则
        Pattern<WarnAmplitudeBean, WarnAmplitudeBean> begin = Pattern.<WarnAmplitudeBean>begin("begin")
                .where(new SimpleCondition<WarnAmplitudeBean>() {
                    @Override
                    public boolean filter(WarnAmplitudeBean value) throws Exception {
                        return value.getAmplitude().compareTo(amplitudeThreshold) == 1;
                    }
                });
        //7.查询数据
        PatternStream<WarnAmplitudeBean> pattern = CEP.pattern(resDate.keyBy(WarnAmplitudeBean::getSecCode), begin);
        pattern.select(new PatternSelectFunction<WarnAmplitudeBean, Object>() {
            @Override
            public Object select(Map<String, List<WarnAmplitudeBean>> pattern) throws Exception {
                List<WarnAmplitudeBean> list = pattern.get("begin");
                //邮件告警
                if(list != null && list.size() >0){
                    MailSend.send("振幅数据实时预警，详情如下："+list.toString());
                }
                return list;
            }
        }).print();

    }
}
